package com.hkbigdata.window;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.stream.Stream;

/**
 * @author liuanbo
 * @creat 2024-04-25-14:36
 * @see 2194550857@qq.com
 */
public class Flink02_ProcessTime_Sliding_Windows {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        env.socketTextStream("hadoop102",9999)
                .flatMap((String data, Collector< Tuple2<String,Integer>> out)->{
                    String[] split = data.split(",");
                    Stream<String> stream = Arrays.stream(split);
                    stream.forEach(word->out.collect(Tuple2.of(word,1)));
                }).returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(data->data.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                .sum(1)
                .print();
        env.execute();

    }
}
